package textToHbase.tyouiifan;

import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

public class HbaseAccess {

	HTable hTable;
	Result r;
	HBaseAdmin admin;
	String tableName;

	public HbaseAccess(String tableName) {
		HBaseConfiguration conf = new HBaseConfiguration();
		conf.addResource(new Path("hbase-site.xml"));
		hTable = null;
		try {
			admin = new HBaseAdmin(conf);
			if (admin.isTableAvailable(tableName)) {
				System.out.println("table exist.");
			} else {
				HTableDescriptor table = new HTableDescriptor(tableName);
				admin.createTable(table);
				System.out.println("created table.");
			}
			this.tableName = tableName;
			hTable = new HTable(conf, tableName);
		} catch (MasterNotRunningException e) {
			// TODO 自動生成された catch ブロック
			e.printStackTrace();
		} catch (ZooKeeperConnectionException e) {
			// TODO 自動生成された catch ブロック
			e.printStackTrace();
		} catch (IOException e) {
			// TODO 自動生成された catch ブロック
			e.printStackTrace();
		}
	}

	public String get(String key, String family, String colum) {
		try {
			r = hTable.get(new Get(Bytes.toBytes(key)));
		} catch (IOException e) {
			// TODO 自動生成された catch ブロック
			e.printStackTrace();
		}
		return Bytes.toString(r.getValue(Bytes.toBytes(family),
				Bytes.toBytes(colum)));
	}

	public KeyValue[] get(String key) {
		try {
			r = hTable.get(new Get(Bytes.toBytes(key)));
			KeyValue[] raw = r.raw();
			return raw;
		} catch (IOException e) {
			// TODO 自動生成された catch ブロック
			e.printStackTrace();
		}
		return null;
	}

	public void put(KeyValue kv) {
		Put p = new Put(kv.getRow());
		// System.out.println(Bytes.toString(kv.getFamily()));
		// System.out.println(Bytes.toString(kv.getRow()));
		// System.out.println(Bytes.toString(kv.getValue()));

		try {
			HTableDescriptor desc = admin.getTableDescriptor(Bytes
					.toBytes(tableName));
			if (desc.getFamily(kv.getFamily()) != null) {
				System.out.println("family name exist");
			} else {
				admin.disableTable(tableName);
				HColumnDescriptor column = new HColumnDescriptor(kv.getFamily());
				admin.addColumn(tableName, column);
				admin.enableTable(tableName);
			}S
			p.add(kv);
			hTable.put(p);
		} catch (IOException e) {
			// TODO 自動生成された catch ブロック
			e.printStackTrace();
		}
	}

	public void put(String key, String family, String colum, String value) {
		KeyValue kv = new KeyValue(Bytes.toBytes(key), Bytes.toBytes(family),
				Bytes.toBytes(colum), Bytes.toBytes(value));
		put(kv);
	}

	// public void put(String key, String value) {
	// KeyValue kv = new KeyValue(Bytes.toBytes(key),
	// HConstants.EMPTY_BYTE_ARRAY, HConstants.EMPTY_BYTE_ARRAY,
	// Bytes.toBytes(value));
	// put(kv);
	// }
}
